use std::{
    collections::{HashMap, HashSet, LinkedList},
    sync::{atomic::AtomicI64, Arc, RwLock},
};

use async_trait::async_trait;
use common::{
    base::id_generator,
    protocol::{get_version, protocol_v1::ProtocolV1Reader, XID},
    service::ServiceKey,
    utils::time_utils,
};
use semver::Version;
use tokio::sync::{
    mpsc::{self, UnboundedReceiver, UnboundedSender},
    Mutex,
};
use tracing::info;

use crate::{
    airport::{airport::Airport, channel::Channel},
    protocol::MsgPriority,
};

use super::{local_service::LocalService, service::Service};

enum MessageSignal {
    Normal(XID),                      // 处理信息
    Handle,                           // 单纯通知处理信息，无消息ID
    Exit,                             // 退出
    Redirect(Arc<Box<LocalService>>), // 转发 信息到本地 service
}

impl MessageSignal {
    fn is_empty(&self) -> bool {
        match self {
            MessageSignal::Normal(xid) => xid.is_empty(),
            _ => false,
        }
    }
}

pub struct RemoteService {
    airport: Arc<Box<Airport>>,

    dxc_name: String,

    dxc_version: Version,

    service_name: String,

    service_id: AtomicI64,

    online_time: AtomicI64,
    /**
     * 通知有消息
     */
    message_receiver: Mutex<UnboundedReceiver<MessageSignal>>,
    /**
     * 消息借
     */
    message_sender: UnboundedSender<MessageSignal>,

    channel_id_set: RwLock<HashSet<i64>>,

    wait_ack_message_id_map: Mutex<HashMap<i64, XID>>,
}

impl RemoteService {
    pub fn new(
        dxc_id: i64,
        dxc_name: String,
        dxc_version: Version,
        service_name: String,

        airport: Arc<Box<Airport>>,
    ) -> Box<RemoteService> {
        let (sender, receiver) = mpsc::unbounded_channel::<MessageSignal>();
        Box::new(RemoteService {
            service_id: AtomicI64::new(id_generator::gen_service_id(dxc_id)),
            airport,
            wait_ack_message_id_map: Mutex::new(HashMap::new()),
            dxc_name: dxc_name,
            dxc_version: dxc_version,
            service_name: service_name,
            online_time: AtomicI64::new(0),
            message_receiver: Mutex::new(receiver),
            message_sender: sender,
            channel_id_set: RwLock::new(HashSet::new()),
        })
    }

    pub fn add_channel_id(&self, channel_id: i64) {
        let mut channel_set = self.channel_id_set.write().unwrap();

        channel_set.insert(channel_id);

        let _ = self.message_sender.send(MessageSignal::Handle);

    }

    pub fn has_channel(&self) -> bool {
        let channel_set = self.channel_id_set.read().unwrap();
        channel_set.is_empty()
    }
    //
    pub fn clear_channel(&self) {
        let mut channel_set = self.channel_id_set.write().unwrap();

        channel_set.clear();
    }

    pub fn remove_channel_id(&self, channel_id: i64) {
        let mut channel_set = self.channel_id_set.write().unwrap();
        channel_set.remove(&channel_id);
    }

    fn get_one_channel_id(&self) -> Option<i64> {
        let channel_set = self.channel_id_set.read().unwrap();
        if channel_set.is_empty() {
            None
        } else {
            Some(*channel_set.iter().next().unwrap())
        }
    }

    async fn get_channel(&self) -> Option<Arc<Box<dyn Channel>>> {
        let channel_id = self.get_one_channel_id();
        if channel_id.is_none() {
            info!("无 channel ...");
            return None;
        }
        self.airport
            .get_channel_manager()
            .get_channel(channel_id.unwrap())
            .await
    }

    pub fn update_id(&self, service_id: i64) {
        self.service_id
            .store(service_id, std::sync::atomic::Ordering::Release);
    }

    pub fn redirect_message(&self, local_service: Arc<Box<LocalService>>) {
        let _ = self
            .message_sender
            .send(MessageSignal::Redirect(local_service));
    }

    pub async fn ack_message(&self, message_id: i64) -> bool {
        // 删除消息
        let xid = {
            let mut wait_ack_message_id_map = self.wait_ack_message_id_map.blocking_lock();

            wait_ack_message_id_map.remove(&message_id)
        };

        if xid.is_none() {
            return false;
        }

        self.airport
            .get_message_manager()
            .delete_message(xid.unwrap())
            .await;

        true
    }

    async fn handle_message(&self) {
        info!("远程 service 处理消息...");
        let mut msg_xid_list: LinkedList<XID> = LinkedList::new();
        let mut message_receiver = self.message_receiver.lock().await;
        while let Some(signal) = message_receiver.recv().await {
            //匹配消息并处理
            match signal {
                MessageSignal::Normal(xid) => {
                    msg_xid_list.push_back(xid);
                }
                MessageSignal::Exit => {
                    break;
                }
                MessageSignal::Redirect(local_service) => {
                    for xid in msg_xid_list {
                        //
                        let message = self.airport.get_message_manager().get_msg(&xid).await;

                        if message.is_none() {
                            continue;
                        }

                        let (message, _) = message.unwrap();

                        local_service.message_in(message.as_ref());
                    }
                    break;
                }
                _ => {}
            };
            let channel = self.get_channel().await;
            if channel.is_none() {
                info!("{:?} 没有 channel...", self.get_service_key());
                continue;
            }
            let mut channel_ins = channel.unwrap();
            let message_manager = self.airport.get_message_manager();
            loop {
                let xid = msg_xid_list.pop_front();
                if xid.is_none() {
                    break;
                }
                let xid = xid.unwrap();
                let message = message_manager.get_msg(&xid).await;
                if message.is_none() {
                    continue;
                }
                let (message, _) = message.unwrap();

                // 发送消息
                let is_succeed = channel_ins.send_message(message.as_ref()).await;
                if !is_succeed {
                    msg_xid_list.push_back(xid);

                    self.remove_channel_id(channel_ins.id());

                    let channel = self.get_channel().await;
                    if channel.is_none() {
                        break;
                    }
                    channel_ins = channel.unwrap();

                    continue;
                }
                //
                {
                    // 插入等待的节点
                    let mut wait_ack_message_id_map = self.wait_ack_message_id_map.lock().await;
                    wait_ack_message_id_map.insert(xid.request_id, xid);
                }

                if msg_xid_list.is_empty() {
                    break;
                }
            }
        }
        message_receiver.close();
    }

    pub fn message_in(&self, message: &Vec<u8>) -> bool {
        let protocol_version = get_version(message);

        match protocol_version {
            1 => {
                let reader = ProtocolV1Reader::new(message);
                let xid = reader.xid();
                // 发送消息
                let result = self.message_sender.send(MessageSignal::Normal(xid));
                result.is_ok()
            }
            _ => false,
        }
    }

    pub fn start(self: Arc<Box<Self>>) {
        self.online_time.store(
            time_utils::cur_timestamp(),
            std::sync::atomic::Ordering::Relaxed,
        );

        let clone_self = self.clone();
        tokio::spawn(async move {
            clone_self.handle_message().await;
        });
    }
}

#[async_trait]
impl Service for RemoteService {
    /**
     * id
     */
    fn id(&self) -> i64 {
        self.service_id.load(std::sync::atomic::Ordering::Acquire)
    }

    /**
     * 获取版本号, 返回的是序列化后的 protobuf 格式的字符串
     */

    fn version(&self) -> Version {
        self.dxc_version.clone()
    }

    fn name(&self) -> &str {
        &self.service_name
    }
    /**
     * 初始化, 此时还不能接受任务
     */
    fn init(&self, _config: &str) {}
    /**
     * 可以开始处理任务
     */

    fn dispatch_message_with_pri(&self, _pri: MsgPriority, _xid: XID) {}

    /**
     * 主动关闭
     */
    async fn stop(&self) {
        let _ = self.message_sender.send(MessageSignal::Exit);
        self.message_sender.closed().await;
    }

    /**
     * 终结
     */
    fn finalize(&self) {}

    fn get_service_key(&self) -> ServiceKey {
        ServiceKey::new(
            self.dxc_name.clone(),
            self.dxc_version.to_string(),
            self.service_name.clone(),
        )
    }
    /**
     * 执行并行任务
     */
    fn execute_parallel_task(&self, task_id: i64) -> bool {
        false
    }
    /**
     * 禁止注册
     */
    fn disable_register(&self) {}
    /**
     * 判断是否需要注册
     */
    fn is_register(&self) -> bool {
        true
    }
    /**
     * 设置私有
     */
    fn set_private(&self, private: bool) {}
    /**
     * 是否是公有的
     */
    fn is_public(&self) -> bool {
        false
    }
    /**
     *
     */
    fn set_master_id(&mut self, master_id: i64) {}
    /**
     *
     */
    fn master_id(&self) -> i64 {
        0
    }
    /**
     * 在线时间
     */
    fn oline_time(&self) -> i64 {
        self.online_time.load(std::sync::atomic::Ordering::Relaxed)
    }
    /**
     *
     */
    fn set_token_id(&self, token_id: i64) {}
    /**
     *
     */
    fn token_id(&self) -> i64 {
        0
    }
}
